package etcd_watch_origin

import (
	"context"
	"errors"
	"fmt"
	"go.etcd.io/etcd/client/v3"
	"sync/atomic"
	"time"
)

// ObserverFunc 监视键值改变的函数
type ObserverFunc func(data *Data)

var WatcherIsRun = errors.New("这个watcher正在运行")

// Watcher 观察者
type Watcher struct {
	*options
	// 运行标志
	run int32
	// 记录版本
	recordVer map[string]int64
}

func newWatcher(opts *options) *Watcher {
	w := &Watcher{
		options: opts,
	}
	return w
}

// NewWatcher 创建一个观察者
func NewWatcher(etcdClient *clientv3.Client, opts ...Option) *Watcher {
	opt := newOptions(etcdClient)
	opt.apply(opts...)
	return newWatcher(opt)
}

// Watch 开始监视
// 如果你需要监视一个key前缀, 请设置 clientv3.WithPrefix() 选项
// 如果你想要知道key在改变之前的数据, 请设置 clientv3.WithPrevKV() 选项
func (w *Watcher) Watch(key string, fn ObserverFunc, opts ...clientv3.OpOption) error {
	if fn == nil {
		panic("ObserverFunc is nil")
	}

	if !atomic.CompareAndSwapInt32(&w.run, 0, 1) {
		return WatcherIsRun
	}

	w.recordVer = make(map[string]int64)

	var err error
	for w.IsRun() {
		select {
		case <-w.ctx.Done():
			fmt.Println("watch cancelled")
			return nil
		default:
			err = w.watch(w.ctx, key, fn, opts...)
			if !w.IsRun() {
				break
			}
			if err != nil {
				time.Sleep(w.retryWaitTime)
			}
		}

	}
	return nil
}

// IsRun 是否运行中
func (w *Watcher) IsRun() bool {
	return atomic.LoadInt32(&w.run) == 1
}

func (w *Watcher) watch(ctx context.Context, key string, fn ObserverFunc, opts ...clientv3.OpOption) error {
	fmt.Println("监听key:", key)
	ch := w.client.Watch(ctx, key, opts...)
	for v := range ch {
		if v.Err() != nil {
			return v.Err()
		}
		for _, e := range v.Events {
			data := newData(e)
			if data.ModRevision == w.recordVer[data.Key] {
				continue // 过滤掉重复的
			}

			w.recordVer[data.Key] = data.ModRevision
			fn(data)
		}
	}
	return nil
}
